import com.alibaba.fastjson.JSON;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import io.debezium.embedded.ConvertingEngineBuilder;
import io.debezium.embedded.ConvertingEngineBuilderFactory;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.planner.expressions.converter.CustomizedConvertRule;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.json.DecimalFormat;
import org.apache.kafka.connect.json.JsonConverterConfig;
import org.roaringbitmap.RoaringBitmap;

import java.io.Serializable;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.HashMap;

public class Demo13 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(2000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");


        HashMap<String, Object> config = new HashMap<>();
        config.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, DecimalFormat.NUMERIC.name());
        // 创建mysql cdc source 对象
        MySqlSource<String> source = MySqlSource.<String>builder()
                .username("root")
                .password("ABC123.abc123")
                .hostname("doitedu01")
                .port(3306)
                .databaseList("doit50")
                .tableList("doit50.oms_order")
                .deserializer(new JsonDebeziumDeserializationSchema(false, config))
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "mysql-cdc");

        SingleOutputStreamOperator<CdcBean> beanStream = stream.map(json -> JSON.parseObject(json, CdcBean.class));


        SingleOutputStreamOperator<CdcBean> filtered = beanStream.filter(new FilterFunction<CdcBean>() {
            @Override
            public boolean filter(CdcBean bean) throws Exception {
                LocalDateTime createTime = bean.after.getCreate_time();
                LocalDateTime paymentTime = bean.after.getPayment_time();
                LocalDateTime confirmTime = bean.after.getConfirm_time();
                LocalDateTime deliveryTime = bean.after.getDelivery_time();


                LocalDate curDate = LocalDate.now();

                if (createTime != null && createTime.toLocalDate().equals(curDate)) return true;
                if (paymentTime != null && paymentTime.toLocalDate().equals(curDate)) return true;
                if (confirmTime != null && confirmTime.toLocalDate().equals(curDate)) return true;
                if (deliveryTime != null && deliveryTime.toLocalDate().equals(curDate)) return true;

                return false;
            }
        });

        SingleOutputStreamOperator<Pair> pairStream = beanStream.process(new ProcessFunction<CdcBean, Pair>() {
            @Override
            public void processElement(CdcBean bean, ProcessFunction<CdcBean, Pair>.Context ctx, Collector<Pair> out) throws Exception {

                OrderBean before = bean.before;
                OrderBean after = bean.after;
                String op = bean.getOp();

                LocalDateTime createTime = after.getCreate_time();
                LocalDateTime paymentTime = after.getPayment_time();
                LocalDateTime confirmTime = after.getConfirm_time();
                LocalDateTime deliveryTime = after.getDelivery_time();

                BigDecimal totalAmount = after.getPay_amount();
                int status = after.status;
                BigDecimal payAmount = after.getPay_amount();
                int id = after.getId();


                LocalDate curDate = LocalDate.now();

                switch (op) {
                    case "r":
                    case "c":
                        // 总
                        if (createTime.toLocalDate().equals(curDate)) {
                            out.collect(new Pair(1, 1, totalAmount, id));
                        }

                        // 待支付
                        if (createTime.toLocalDate().equals(curDate) && status == 0) {
                            out.collect(new Pair(2, 1, totalAmount, id));
                        }
                        // 支付
                        if (paymentTime != null && paymentTime.toLocalDate().equals(curDate) && status == 1) {
                            out.collect(new Pair(3, 1, payAmount, id));
                        }
                        // 发货
                        if (deliveryTime != null && deliveryTime.toLocalDate().equals(curDate) && status == 2) {
                            out.collect(new Pair(4, 1, payAmount, id));
                        }
                        // 完成
                        if (confirmTime != null && confirmTime.toLocalDate().equals(curDate) && status == 3) {
                            out.collect(new Pair(5, 1, payAmount, id));
                        }
                        break;
                    case "u":
                        // 未支付->已支付
                        if (before.status == 0 && status == 1) {
                            out.collect(new Pair(2, -1, before.pay_amount.negate(), id));
                            out.collect(new Pair(3, 1, after.pay_amount, id));
                        }
                        // 已支付->已发货
                        else if (before.status == 1 && status == 2) {
                            out.collect(new Pair(4, 1, after.pay_amount, id));
                        }
                        // 已发货->已确认
                        else if (before.status == 2 && status == 3) {
                            out.collect(new Pair(5, 1, after.pay_amount, id));
                        }
                        break;
                    default:

                }


            }
        });


        SingleOutputStreamOperator<String> resultStream = pairStream.keyBy(p -> 0)
                .process(new KeyedProcessFunction<Integer, Pair, String>() {

                    MapState<Integer, RoaringBitmap> cntState;
                    ValueState<Result> r;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        cntState = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, RoaringBitmap>("cnt", Integer.class, RoaringBitmap.class));
                        r = getRuntimeContext().getState(new ValueStateDescriptor<Result>("r", Result.class));
                    }

                    @Override
                    public void processElement(Pair pair, KeyedProcessFunction<Integer, Pair, String>.Context ctx, Collector<String> out) throws Exception {

                        int oid = pair.id;
                        int idType = pair.idType;
                        BigDecimal amt = pair.amt;
                        int idx = pair.idx;


                        RoaringBitmap oidBm = cntState.get(idx);
                        if (oidBm == null) {
                            oidBm = RoaringBitmap.bitmapOf();
                        }
                        if (idType == 1) {
                            oidBm.add(oid);
                        } else {
                            oidBm.remove(oid);
                        }
                        cntState.put(idx, oidBm);


                        Result result = r.value();
                        if (result == null) {
                            BigDecimal zero = BigDecimal.ZERO;
                            result = new Result(0,zero,zero,0,zero,0,zero,0,zero,0,zero);
                        }

                        switch (idx) {
                            case 1:
                                result.orderTotalAmt = result.orderTotalAmt.add(amt);
                                break;
                            case 2:
                                result.toPayAmt = result.toPayAmt.add(amt);
                                break;
                            case 3:
                                result.payedAmt = result.payedAmt.add(amt);
                                break;
                            case 4:
                                result.delieveredAmt = result.delieveredAmt.add(amt);
                                break;
                            case 5:
                                result.confirmedAmt = result.confirmedAmt.add(amt);
                                break;
                            default:
                        }
                        r.update(result);


                        result.orderTotalCnt = cntState.get(1) != null ? cntState.get(1).getCardinality() : 0;
                        result.toPayCnt = cntState.get(2) != null ? cntState.get(2).getCardinality() : 0;
                        result.payedCnt = cntState.get(3) != null ? cntState.get(3).getCardinality() : 0;
                        result.delieveredCnt = cntState.get(4) != null ? cntState.get(4).getCardinality() : 0;
                        result.confirmedCnt = cntState.get(5) != null ? cntState.get(5).getCardinality() : 0;

                        out.collect(JSON.toJSONString(result));
                    }
                }).setParallelism(1);


        resultStream.print().setParallelism(1);

        env.execute();


    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Result implements Serializable{
        private int orderTotalCnt;
        private BigDecimal orderTotalAmt;
        private BigDecimal orderPayAmt;
        private int toPayCnt;
        private BigDecimal toPayAmt;
        private int payedCnt;
        private BigDecimal payedAmt;
        private int delieveredCnt;
        private BigDecimal delieveredAmt;
        private int confirmedCnt;
        private BigDecimal confirmedAmt;
    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Pair implements Serializable {
        private int idx;
        private int idType;
        private BigDecimal amt;
        private int id;
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class OrderBean implements Serializable {
        private int id;
        private int status;
        private BigDecimal total_amount;
        private BigDecimal pay_amount;
        private LocalDateTime create_time;
        private LocalDateTime payment_time;
        private LocalDateTime delivery_time;
        private LocalDateTime confirm_time;

    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class CdcBean implements Serializable {
        private OrderBean before;
        private OrderBean after;
        private String op;

    }
}
